#include "XThread.h"
#include "XTask.h"
#include "LogWrapper.h"
#include <thread>
#include <event2/event.h>
#include <event2/util.h>
#ifdef _WIN32
#else
#inlude <unistd.h>
#endif

static void notifyHandle(evutil_socket_t fd, short which, void *arg)
{
	XThread *thread = (XThread *)arg;
	thread->notify(fd, which);
}

void XThread::notify(evutil_socket_t fd, short which)
{
	char buf[2] = { 0 };
#ifdef _WIN32
	int ret = recv(fd, buf, 1, 0);
#else
	int ret = read(fd, buf, 1);
#endif
	if (ret <= 0) {
		return;
	}
    XTask *task = nullptr;
    do 
    {
        std::lock_guard<std::mutex> lock(mMutex);
        if (mTaskQueue.empty()) {
            break;
        }
        task = mTaskQueue.front();
        mTaskQueue.pop_front();
    } while (0);
    LOG_INFO("wakeup thread-{0}:{1}", mId, buf);
    if (task) {
        task->init();
    }
}

void XThread::activate()
{
#ifdef _WIN32
	int ret = send(mNotifySendFd, "c", 1, 0);
#else
	int ret = write(mNotifySendFd, "c", 1);
#endif
	if (ret <= 0) {
		LOG_ERROR("XThread::activate() failed ");
	}
}

void XThread::addTask(XTask * task)
{
    do {
        if (!task) {
            break;
        }
        task->mBase = mEventBase;
        task->mThreadId = mId;
        std::lock_guard<std::mutex> lock(mMutex);
        mTaskQueue.push_back(task);
    } while (0);
}

XThread::XThread()
{

}


XThread::~XThread()
{
}

void XThread::start()
{
	setup();
	std::thread th(&XThread::main, this);
	th.detach();
}

bool XThread::setup()
{
#ifdef _WIN32
	evutil_socket_t fds[2];
	if (evutil_socketpair(AF_INET, SOCK_STREAM, 0, fds) < 0) {
		LOG_ERROR("evutil_socketpair failed!");
		return false;
	}
	evutil_make_socket_nonblocking(fds[0]);
	evutil_make_socket_nonblocking(fds[1]);
#else
	int fds[2];
	if (pipe(fds)) {
		cerr << "pipe failed" << endl;
		return false;
	}
#endif
	mNotifySendFd = fds[1];
	event_config *ev_config = event_config_new();
	event_config_set_flag(ev_config, EVENT_BASE_FLAG_NOLOCK);
	mEventBase = event_base_new_with_config(ev_config);
	event_config_free(ev_config);
	if (!mEventBase) {
		LOG_ERROR("event_base_new_with_config failed");
		return false;
	}
	event *ev = event_new(mEventBase, fds[0], EV_READ | EV_PERSIST, notifyHandle, this);
	event_add(ev, 0);
	return true;
}

void XThread::main()
{
	LOG_DEBUG("thread id {} start", mId);
	event_base_dispatch(mEventBase);
	event_base_free(mEventBase);
    LOG_DEBUG("thread id {} end", mId);
}